package kafkaBus;

import java.util.Date;
import java.util.Properties;
import java.util.Random;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

/**
 * Created by Administrator on 2017/4/13 0013.
 */
public class SendDataToKafka {


    public static void main(String[] args) {
        long events = Long.parseLong("10");
        Random rnd = new Random();

        Properties props = new Properties();
//        props.put("metadata.broker.list","192.168.188.133:9092,192.168.188.134:9092,192.168.188.136:9092");
        props.put("metadata.broker.list","slave-01:9092,slave-02:9092,slave-03:9092");
        props.put("serializer.class", "kafka.serializer.StringEncoder");
        /*props.put("partitioner.class", "example.producer.SimplePartitioner");*/
        props.put("request.required.acks", "1");

        ProducerConfig config = new ProducerConfig(props);
        Producer<String, String> producer = new Producer<String, String>(config);

        for (long nEvents = 0; nEvents < events; nEvents++) {
            long runtime = new Date().getTime();
            String ip = "192.168.2." + rnd.nextInt(255);
            String msg = runtime + ",www.example.com," + ip;
            KeyedMessage<String, String> data = new KeyedMessage<String, String>("page_visits", ip, msg);
            producer.send(data);
        }
        main(args);
        producer.close();
    }

}
